package com.it.stream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @author code1997
 */
public class ConnectStreamTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        DataStreamSource<Integer> source1 = executionEnvironment.fromElements(1, 2, 3);
        DataStreamSource<Long> source2 = executionEnvironment.fromElements(4L, 5L, 6L);
        source1.connect(source2).map(new CoMapFunction<Integer, Long, String>() {
            @Override
            public String map1(Integer value) throws Exception {
                return "Integer type: " + value;
            }

            @Override
            public String map2(Long value) throws Exception {
                return "Long type: " + value;
            }
        }).print("result");
        executionEnvironment.execute();
    }
}
